package com.dongdongshop.consumer;

import com.alibaba.dubbo.config.annotation.Reference;
import com.dongdongshop.pojo.TbIntegral;
import com.dongdongshop.pojo.TbOrder;
import com.dongdongshop.pojo.TbUser;
import com.dongdongshop.service.IntegralService;
import com.dongdongshop.service.OrderService;
import com.dongdongshop.service.UserService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

@Component
@RocketMQMessageListener(topic = "IntegralTopic", consumerGroup = "ddshop-integral")
public class IntegerConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Resource
    private OrderService orderService;

    @Reference
    private UserService userService;

    @Resource
    private IntegralService insertIntegral;

    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                messageExt.getBody();
                messageExt.getMsgId();
                try {
                    String body = new String(messageExt.getBody(),"utf-8");
                    String[] split = body.split(",");
                    TbIntegral tbIntegral = new TbIntegral();
                    //根据母订单查询订单（回去用户id）
                    List<TbOrder> orderList = (List<TbOrder>)orderService.getOrdersByParenOrderId(split[0]);
                    tbIntegral.setUserId(orderList.get(0).getUserId());
                    tbIntegral.setUpdateTime(new Date());
                    Double payment = Double.parseDouble(split[1]);
                    Long integral = new Double(payment / 10).longValue();
                    tbIntegral.setIntegralNum(integral);

                    TbUser user = userService.getUserByName(orderList.get(0).getUserId());
                    user.setPoints(integral.intValue() + user.getPoints());
                    //根据用户积分判断 用户等级(1-100 青铜；101-1000 白银;1001-3000 铂金;3001-5000 钻石)
                    if (user.getPoints() > 0 && user.getPoints() <= 100){
                        user.setUserLevel(1);
                    }else if (user.getPoints() > 100 && user.getPoints() <= 1000){
                        user.setUserLevel(2);
                    }else if (user.getPoints() > 1000 && user.getPoints() <= 3000){
                        user.setUserLevel(3);
                    }else if (user.getPoints() > 3000){
                        user.setUserLevel(4);
                    }
                    userService.updateUser(user);
                    //落库
                    tbIntegral.setUserLevel(user.getUserLevel().toString());
                    insertIntegral.insertIntegral(tbIntegral);
                    System.out.println("用户增加积分消息消费成功:" + body);
            }catch (Exception e){
                    System.out.println("用户增加积分消息消费失败");
                    if (messageExt.getReconsumeTimes() == 2){
                        //如果重试次数等于2，业务数据保存的redis，并且停止重试
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }else {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
